package client

import (
	"encoding/json"
	"errors"
	"fmt"
	"gark-elastic-job/common"
	"sync"
	"time"

	logger "github.com/sirupsen/logrus"
)

type GarkJobClient struct {
	// properties
	ip           string
	id           string
	state        RegistryState
	exit         chan bool
	running      bool
	group        string
	clientPath   string
	snapshotPath string
	jobs         map[string]Job

	// components
	etcd              *Etcd
	txResponse        *TxResponse
	snapshotProcessor *SnapshotProcessor
	locker            *sync.RWMutex
}

func NewGarkJobClient(id, group, ip string, etcd *Etcd) (*GarkJobClient, error) {
	err := validate(id, group, ip)
	if err != nil {
		return nil, err
	}
	return &GarkJobClient{
		etcd:         etcd,
		jobs:         make(map[string]Job),
		state:        UnRegistered,
		exit:         make(chan bool),
		group:        group,
		clientPath:   fmt.Sprintf(RegistryPath, group, id),
		snapshotPath: fmt.Sprintf(SnapshotPath, group, id),
		ip:           ip,
		id:           id,
		locker:       &sync.RWMutex{},
	}, nil
}

func (c *GarkJobClient) registerNode() {
RETRY:
	if c.state == Registered {
		logger.Info("the client has already registered")
		return
	}
	clientData := common.ClientMetaData{
		ID:         c.id,
		Group:      c.group,
		IP:         c.ip,
		CreateTime: time.Now(),
	}

	value, err := json.Marshal(clientData)
	if err != nil {
		logger.Error("failed to encode client metadata")
	}

	// TODO 这里应该加入重试逻辑 - 官方不建议使用 goto
	txResp, err := c.etcd.TxKeepaliveWithTTL(c.clientPath, string(value), 10)
	if err != nil {
		logger.Errorf("failed to register client to the path: %s, err: %v", c.clientPath, err)
		time.Sleep(3 * time.Second)
		goto RETRY
	}

	if !txResp.Success {
		logger.Errorf("failed to register client to the path: %s", c.clientPath)
		time.Sleep(3 * time.Second)
		goto RETRY
	}

	logger.Infof("success to register client to the path: %s", c.clientPath)
	c.state = Registered
	c.txResponse = txResp

	select {
	case <-c.txResponse.StateChan:
		c.state = UnRegistered
		logger.Warnf("client lost the signal to register to the path: %s", c.clientPath)
		goto RETRY
	}

}

// 客户端不断检查自己的目录下面有没有要执行的任务
// 这种模式是采用 pull的方式进行交互（类似于不断的去检查某个节点）
func (c *GarkJobClient) lookup() {

	for {
		keys, values, err := c.etcd.GetWithPrefixLimit(c.snapshotPath, 50)
		if err != nil {
			logger.Debugf("client load job snapshot error:%v", err)
			time.Sleep(time.Second * 3)
			continue
		}

		if c.state == UnRegistered {
			logger.Infof("client has not yet registered, will be sleep 3 seconds to retry")
			time.Sleep(time.Second * 3)
			continue
		}

		if len(keys) == 0 || len(values) == 0 {
			logger.Infof("client load job snapshot empty for the path: %s", c.clientPath)
			time.Sleep(time.Second * 3)
			continue
		}

		for i := 0; i < len(values); i++ {
			key := keys[i]

			if c.state == UnRegistered {
				time.Sleep(time.Second * 3)
				break
			}

			logger.Infof("start delete key: %s snapshot to avoid repeat", key)
			if err := c.etcd.Delete(string(key)); err != nil {
				logger.Warnf("client failed to delete snapshot from path: %s, err: %v", c.clientPath, err)
				continue
			}

			value := values[i]
			if len(value) == 0 {
				logger.Warnf("client for path %s found the key but value is nil", c.clientPath)
				continue
			}

			snapshot := &common.GarkJobSnapshot{}
			err := json.Unmarshal(value, snapshot)
			if err != nil {
				logger.Errorf("decode snapshot failed err: %v", err)
				continue
			}
			c.snapshotProcessor.pushSnapshot(snapshot)
		}
	}
}

func (c *GarkJobClient) PushJob(name string, job Job) {
	c.locker.Lock()
	defer c.locker.Unlock()

	if _, ok := c.jobs[name]; ok {
		logger.Warnf("job %s: %v already exists", name, job)
		return
	}
	c.jobs[name] = job
}

func (c *GarkJobClient) Bootstrap() {
	c.snapshotProcessor = NewSnapshotProcessor(c.id, c.group, c.etcd)

	// 添加任务要早于开启goroutine
	c.addJobs()

	go c.registerNode()
	go c.lookup()

	c.running = true
	<-c.exit
	c.running = false
}

func (c *GarkJobClient) Stop() {
	if c.running {
		c.exit <- true
	}
	return
}

func (c *GarkJobClient) addJobs() {
	if len(c.jobs) == 0 {
		return
	}
	for name, job := range c.jobs {
		c.snapshotProcessor.PushJob(name, job)
	}
}

func validate(id string, group string, ip string) error {
	if id == "" {
		return errors.New("id can't allow empty")
	}
	if group == "" {
		return errors.New("group can't allow empty")
	}
	if ip == "" {
		return errors.New("ip can't allow empty")
	}
	return nil
}
